package com.its.test.testdemo.kafka;

import cn.hutool.json.JSONUtil;
import com.its.common.kafka.annotation.KafkaHandler;
import com.its.common.kafka.consumer.AbstractKafkaConsumerWorker;
import com.its.test.testdemo.controller.TestObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

/**
 * @Author xiaxp
 * @Date 2021-11-10 20:53
 * @Description
 */
@Slf4j
@ConditionalOnProperty(prefix = "its.kafka.consumer", name = "enabled", havingValue = "true")
@KafkaHandler(bootstrapServers = "localhost:9092", topic = "t-abc", groupId = "g2", workNum = 5)
public class KafkaConsumerWorker extends AbstractKafkaConsumerWorker {
    @Override
    public void handleRecord(ConsumerRecord record) {
        System.err.printf("Thread - "+ Thread.currentThread().getName()
                        + "partition = %d , offset = %d, key = %s, value = %s%n",
                record.partition(), record.offset(), record.key(), record.value());
        String value = (String) record.value();



        TestObject object = JSONUtil.toBean(value, TestObject.class);


        System.out.println(object.getLocalDateTime());
    }
}
